use std::sync::Mutex;

use futures::StreamExt;
use mongodb::{
    bson::{self, doc, oid::ObjectId, Document},
    options::FindOptions,
    Client, Collection,
};
use once_cell::sync::Lazy;
use serde::{Deserialize, Serialize};

use crate::Result;

static TEMP: Lazy<Mutex<Option<Client>>> = Lazy::new(Mutex::default);
struct Collections {
    client: Client,
    umsg: Collection<Document>,
}

static COLLECTIONS: Lazy<Collections> = Lazy::new(|| {
    let client = TEMP.lock().unwrap().take().unwrap();
    Collections {
        umsg: client.database("rmsg").collection("umsg"),
        client,
    }
});

pub fn set_client(client: Client) {
    *TEMP.lock().unwrap() = Some(client);
}

#[allow(dead_code)]
fn client() -> &'static Client {
    &COLLECTIONS.client
}

pub fn umsg_coll() -> &'static Collection<Document> {
    &COLLECTIONS.umsg
}

#[derive(Serialize, Deserialize, Debug)]
pub struct UmsgDb {
    pub _id: ObjectId,
    pub uid: i64,
    pub create_time: i64,
    pub ack_time: i64,
    pub msg_type: UmsgType,
    pub msg: Document,
}

pub async fn insert(uid: i64, msg_type: UmsgType, msg: Document) -> Result<ObjectId> {
    let coll = &COLLECTIONS.umsg;

    let umsg = UmsgDb {
        _id: ObjectId::new(),
        uid,
        create_time: chrono::Local::now().timestamp(),
        ack_time: 0,
        msg_type,
        msg,
    };

    let doc = bson::to_document(&umsg)?;
    let result = coll.insert_one(doc, None).await?;
    result.inserted_id.as_object_id().clone().ok_or_else(|| {
        anyhow!(
            "insert umsg uid:{} type:{:?} msg:{:?} --- as_object_id error",
            umsg.uid,
            umsg.msg_type,
            umsg.msg
        )
    })
}

pub async fn list_no_ack_messages(uid: i64) -> Result<Vec<UmsgDb>> {
    let coll = &COLLECTIONS.umsg;

    let find_options = FindOptions::builder().sort(doc! {"_id": 1}).build();
    let mut cursor = coll
        .find(doc! {"uid": uid, "ack_time": 0}, find_options)
        .await?;

    let mut messages: Vec<UmsgDb> = Vec::new();
    while let Some(doc) = cursor.next().await {
        let doc = doc?;
        messages.push(bson::from_document(doc)?);
    }
    Ok(messages)
}

pub async fn ack_message(uid: i64, object_id: &str) -> Result<()> {
    let coll = &COLLECTIONS.umsg;
    coll.update_one(
        doc! {"_id": ObjectId::parse_str(object_id)?, "uid": uid},
        doc! {"$set": {
            "ack_time": chrono::Local::now().timestamp(),
        }},
        None,
    )
    .await?;
    Ok(())
}

#[derive(Serialize, Deserialize, Copy, Clone, Debug)]
pub enum UmsgType {
    UmsgTest,
}

#[derive(Serialize, Deserialize, Debug)]
pub struct UmsgTest {
    pub order_id: i64,
    pub product_id: String,
}
